package taskpool

import (
	"context"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pkg/errors"
	"testkjcld.bhecard.com/gitlab/kit/esim/log"
)

// ConCurrRun 并行执行任务.
func ConCurrRun(ctx context.Context, logger log.Logger, conCurrency, timeOut int, d Dispatcher, w SubWorker) ([]TaskItemStat, error) {
	var wg sync.WaitGroup
	var workerResult string
	var dispatcherResult string
	var dispatcherCount int64 = 0
	var workerCount int64 = 0

	ctx = logger.ExtendTracerID(ctx)

	logger.Infoc(ctx, "ConCurrRun 开始运行;")
	begin := time.Now()

	// 最少默认值
	if conCurrency <= 0 {
		conCurrency = defaultPoolSize
	}

	// 最少等待5秒后，退出
	if timeOut < 5 {
		timeOut = 5
	}

	// worker 分发通道
	taskChan := make(chan interface{})
	// 接收worker 处理结果
	statChan := make(chan *TaskItemStat, conCurrency)
	// 状态统计
	statList := make([]TaskItemStat, 0)
	// error列表
	errorChan := make(chan error, conCurrency+1)

	nCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	// 管理worker callback
	wg.Add(1)
	go func() {
		defer wg.Done()
		callbackFunc(ctx, logger, conCurrency, statChan, statList, errorChan, cancel, &workerResult)
		logger.Infoc(ctx, "callbackFunc return;")
	}()

	// 启动worker 接受task
	for i := 1; i <= conCurrency; i++ {
		wg.Add(1)
		logger.Infoc(nCtx, "开始启动worker [%d];", i)
		// 启动work 接收子任务
		go func(idx int) {
			defer wg.Done()
			workFunc1(nCtx, logger, w, taskChan, idx, timeOut, &workerCount, statChan)
			logger.Infoc(nCtx, "workFunc1 return;")
		}(i)
	}

	// 开始启动 dispatcher
	wg.Add(1)
	go func() {
		defer wg.Done()
		dispatcherFunc(nCtx, logger, d, taskChan, cancel, errorChan, &dispatcherCount, &dispatcherResult)
		logger.Infoc(nCtx, "dispatcherFunc return;")
	}()

	// goroutine 执行结束
	wg.Wait()

	logger.Infoc(ctx, "ConCurrRun 执行完成; Worker[%s] Dispatcher[%s] 耗时[%v]", workerResult, dispatcherResult, time.Since(begin))

	if workerResult == Fail || dispatcherResult == Fail || len(errorChan) > 0 {
		errList := chan2List(errorChan)
		logger.Infoc(ctx, "Pool任务执行失败, 分发状态[%s] 执行状态[%s] 子任务执行状态列表[%+v] Error list[%+v]", dispatcherResult, workerResult, statList, errList)
		return statList, errors.Errorf("Pool任务执行失败Error[%s]", errList[0])
	}

	if dispatcherCount != workerCount {
		return statList, errors.Errorf("Pool任务执行失败,分发任务[%d]和执行任务[%d]不匹配 子任务执行状态列表[%+v];", dispatcherCount, workerCount, statList)
	}

	return statList, nil
}

// ConCurrRunList 并行执行任务列表.
func ConCurrRunList(ctx context.Context, logger log.Logger, taskList []interface{}, w SubWorker) ([]TaskItemStat, error) {
	var wg sync.WaitGroup
	var workerResult string
	taskNum := len(taskList)

	ctx = logger.ExtendTracerID(ctx)

	logger.Infoc(ctx, "ConCurrRunList 开始运行;")
	begin := time.Now()

	if len(taskList) == 0 {
		return nil, errors.Errorf("任务列表[%+v]为空;", taskList)
	}

	// 接收worker 处理结果
	statChan := make(chan *TaskItemStat, taskNum)
	statList := make([]TaskItemStat, taskNum)

	errorChan := make(chan error, taskNum)

	nCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	// 管理worker callback
	wg.Add(1)
	go func() {
		defer wg.Done()
		callbackFunc(ctx, logger, taskNum, statChan, statList, errorChan, cancel, &workerResult)
	}()

	// 按照task分配任务
	for i := 0; i < taskNum; i++ {
		wg.Add(1)
		logger.Infoc(nCtx, "分配任务 [%d];", i)
		// 启动并行执行子任务
		go func(idx int) {
			defer wg.Done()
			workFunc(nCtx, logger, w, taskList[idx], idx, statChan)
		}(i)
	}
	// 等待所有子 goroutine 执行结束
	wg.Wait()
	logger.Infoc(ctx, "ConCurrRunList 执行完成; Worker[%s] 子任务执行状态[%+v] 耗时[%v]", workerResult, statList, time.Since(begin))

	if workerResult == Fail || len(errorChan) > 0 {
		errList := chan2List(errorChan)
		logger.Infoc(ctx, "PoolRunList 执行失败; Worker[%s] 子任务执行状态[%+v] errList[%+v] 耗时[%v]", workerResult, statList, errList, time.Since(begin))
		return statList, errors.Errorf("PoolRunList error[%+v]", errList)
	}
	return statList, nil
}

// ConCurrTask 并行执行任务，待所有任务执行完成后，子任务退出；子任务接收、退出 由子任务自行管理.
func ConCurrTask(ctx context.Context, logger log.Logger, conCurrency int, d Dispatcher, w SubLongWorker) ([]WorkerStat, error) {
	var wg sync.WaitGroup
	var workerResult string
	var workerStat []WorkerStat
	var dispatcherResult string
	var dispatcherCount int64 = 0
	var workerCount int64 = 0

	ctx = logger.ExtendTracerID(ctx)

	logger.Infoc(ctx, "ConCurrTask 开始运行;")
	begin := time.Now()

	// 最少默认值
	if conCurrency <= 0 {
		conCurrency = defaultPoolSize
	}

	// worker 分发通道
	taskChan := make(chan interface{})
	// 接收worker 处理结果
	statChan := make(chan *WorkerStat, conCurrency)
	// error列表
	errorChan := make(chan error, conCurrency+1)

	nCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	// 管理worker callback
	wg.Add(1)
	go func() {
		defer wg.Done()
		workerStat, workerResult = workerRunCallback(ctx, logger, conCurrency, statChan, errorChan, cancel)
	}()

	// 启动worker 接受task conCurrency
	for i := 1; i <= conCurrency; i++ {
		wg.Add(1)
		// worker 子任务
		// taskChan 接收子任务
		// error 返回出错信息
		// workerStat返回任务执行结果
		// 任务执行失败 cancel
		go func(idx int) {
			defer wg.Done()
			logger.Infoc(nCtx, "开始启动 workerRun [%d];", idx)
			stat, err := workerRun(nCtx, idx, logger, taskChan, w)
			if err != nil {
				statChan <- stat
				logger.Errorc(ctx, "workerRun: [%d] 执行失败[%s]", idx, err)
				return
			}
			logger.Infoc(ctx, "workerRun: [%d]执行完成[%v]", idx, stat)
			// 统计执行任务数
			atomic.AddInt64(&workerCount, stat.SuccessTask)
			statChan <- stat
			return
		}(i)
	}

	// 开始启动 dispatcher  1
	wg.Add(1)
	go func() {
		defer wg.Done()
		// 分配任务到taskChan
		// 出错通过 errorChan 返回
		// 返回dispatcherCount 任务分配累计总数
		// 返回dispatcherResult 返回分配结果
		// 任务分配失败cancel
		dispatcherFunc(nCtx, logger, d, taskChan, cancel, errorChan, &dispatcherCount, &dispatcherResult)
	}()

	// goroutine 执行结束
	wg.Wait()

	logger.Infoc(ctx, "ConCurrTask: 执行完成; Worker[%s] Dispatcher[%s] 耗时[%v]", workerResult, dispatcherResult, time.Since(begin))

	if workerResult == Fail || dispatcherResult == Fail || len(errorChan) > 0 {
		errList := chan2List(errorChan)
		logger.Infoc(ctx, "ConCurrTask: Pool任务执行失败, 分发状态[%s] 执行状态[%s] 子任务执行状态列表[%+v]", dispatcherResult, workerResult, workerStat)
		return workerStat, errors.Errorf("Pool任务执行失败Error[%s]", errList[0])
	}

	if dispatcherCount != workerCount {
		return workerStat, errors.Errorf("Pool任务执行失败,分发任务[%d]和执行任务[%d]不匹配 子任务执行状态列表[%+v];", dispatcherCount, workerCount, workerStat)
	}

	return workerStat, nil
}
